package net.opentsdb.uid;

import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.Const;
import net.opentsdb.core.TSDB;
import org.hbase.async.GetRequest;
import org.hbase.async.HBaseClient;
import org.hbase.async.KeyValue;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;

/**
 * @Author hujf
 * @Date 2022/1/21 10:45
 * @Description 编码
 */
public class UniqueId {
    private short id_width;
    private HBaseClient client;

    /** Enumerator for different types of UIDS @since 2.0 */
    public enum UniqueIdType {
        METRIC,
        TAGK,
        TAGV
    }

    public UniqueId(HBaseClient client, byte[] uidtable, String kind, short metricsWidth, boolean b) {
        this.table = uidtable;
        this.kind = kind.getBytes(StandardCharsets.UTF_8);
        this.client = client;
        this.id_width = metricsWidth;
    }

    public static byte[] getTSUIDFromKey(byte[] row_key, short metricsWidth, short timestampBytes) {
        int idx = 0;
        int tag_pair_width = TSDB.tagk_width()+TSDB.tagv_width();
        int tags_length = row_key.length - (Const.SALT_WIDTH()+metricsWidth+timestampBytes);

        final byte[] tsuid = new byte[row_key.length-timestampBytes-Const.SALT_WIDTH()];

        for(int i= Const.SALT_WIDTH();i<row_key.length;i++){
            if(i<Const.SALT_WIDTH()+metricsWidth||i>=(Const.SALT_WIDTH()+metricsWidth+timestampBytes)){
                tsuid[idx] =row_key[i];
                idx++;
            }
        }
        return tsuid;
    }

    public short width() {
        return id_width;
    }

    public byte[] getOrCreateId(String metric) {
        try {
            return getIdAsync(metric).joinUninterruptibly();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public Deferred<byte[]> getIdAsync(String metric) {
        byte[] id  =null;
        if(id!=null){
            //incrementCacheHits();
            return Deferred.fromResult(id);
        }
        incrementCacheMiss();

        class GetIdCB implements Callback<byte[],byte[]>{

            @Override
            public byte[] call(byte[] bytes) throws Exception {
                return bytes;
            }
        }

        Deferred<byte[]> d = getIdFromHbase(metric).addCallback(new GetIdCB());
        return d;
    }

    private Deferred<byte[]> getIdFromHbase(String metric) {
        return hbaseGet(metric.getBytes(StandardCharsets.UTF_8),"id".getBytes(StandardCharsets.UTF_8));
    }

    private final byte[] table;
    private final byte[] kind;
    private Deferred<byte[]> hbaseGet(byte[] key, byte[] family) {
        GetRequest get  = new GetRequest(table,key);
        get.family(family).qualifier(kind);
        class GetCB implements Callback<byte[], ArrayList<KeyValue>>{
            @Override
            public byte[] call(ArrayList<KeyValue> row) throws Exception {
                if(row == null||row.isEmpty()){
                    return null;
                }
                return row.get(0).value();
            }
        }
        return client.get(get).addCallback(new GetCB());
    }

    private volatile long cache_misses;
    private void incrementCacheMiss() {
        if(cache_misses>=Long.MAX_VALUE){
            cache_misses = 1;
        }else {
            cache_misses++;
        }
    }

    public byte[] getId(String metric) {
        try {
            return getIdAsync(metric).joinUninterruptibly();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
